package com.alibaba.otter.canal.instance.core;

import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.ha.CanalHAController;
import com.alibaba.otter.canal.parse.ha.HeartBeatHAController;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event;

/**
 * AbstractCanalInstance是CanalInstance的抽象子类，定义了相关字段来维护eventParser、eventSink、eventStore、metaManager的引用。
 *
 * 在start和stop时，按照一定的顺序启动或停止event store、event sink、event parser、meta manager这几个组件，
 * 期间对于event parser的启动和停止做了特殊处理，并没有提供订阅binlog的相关方法。那么如何来订阅binglog数据呢？答案是直接操作器内部组件。
 *
 * AbstractCanalInstance除了负责启动和停止其内部组件，就没有其他工作了。真正获取binlog信息，以及相关元数据维护的逻辑，都是在CanalServerWithEmbedded中完成的。
 * 事实上，设计的角度，既然这些模块是CanalInstance的内部组件，那么相关操作也应该封装在CanalInstance的实现类中，对外部屏蔽，不应该把这些逻辑放到CanalServerWithEmbedded中实现。
 *
 * AbstractCanalInstance中并没有metaManager、eventSink、eventPaser，eventStore这几个组件。这几个组件的实例化是在AbstractCanalInstance的子类中实现的。
 * AbstractCanalInstance有2个子类：CanalInstanceWithSpring和CanalInstanceWithManager。
 */
public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {

    private static final Logger                      logger = LoggerFactory.getLogger(AbstractCanalInstance.class);

    // 在AbstractCanalInstance中，并没有提供方法来初始化这些字段。
    // 这些字段都是protected的，子类可以直接访问，显然这些字段都是在AbstractCanalInstance的子类中进行赋值的。

    protected Long                                   canalId;       // 和manager交互唯一标示
    protected String                                 destination;   // 队列名字
    protected CanalEventStore<Event>                 eventStore;    // 有序队列

    protected CanalEventParser                       eventParser;   // 解析对应的数据信息
    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;     // 链接parse和store的桥接器
    protected CanalMetaManager                       metaManager;   // 消费信息管理器
    protected CanalAlarmHandler                      alarmHandler;  // alarm报警机制
    protected CanalMQConfig                          mqConfig;      // mq的配置

    /**
     * subscribeChange方法，主要是更新一下eventParser中的filter。
     *
     * filter规定了需要订阅哪些库，哪些表。在服务端和客户端都可以设置，客户端的配置会覆盖服务端的配置。
     *
     * 服务端配置：主要是配置instance.properties中的 canal.instance.filter.regex 配置项，
     * http://static.tianshouzhi.com/ueditor/upload/image/20180506/1525620183452065609.png
     *
     * 客户端配置: 客户端在订阅时，调用CanalConnector接口中定义的带有filter参数的subscribe方法重载形式
     */
    @Override
    public boolean subscribeChange(ClientIdentity identity) {
        if (StringUtils.isNotEmpty(identity.getFilter())) { //如果设置了filter
            logger.info("subscribe filter change to " + identity.getFilter());
            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());

            boolean isGroup = (eventParser instanceof GroupEventParser);
            if (isGroup) {
                // 处理group的模式
                List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
                for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                    if(singleEventParser instanceof AbstractEventParser) {
                        ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
                    }
                }
            } else {
                if(eventParser instanceof AbstractEventParser) {
                    ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
                }
            }
        }

        // filter的处理规则
        // a. parser处理数据过滤处理
        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
        // 后续内存版的一对多分发，可以考虑
        return true;
    }

    /**
     * 主要就是启动各个模块。启动顺序为：metaManager—>eventStore—>eventSink—>eventParser。
     *
     * 官方关于instance模块构成的图中，把metaManager放在最下面，说明其是最基础的部分，因此应该最先启动。
     * eventParser依赖于eventSink，需要把自己解析的binlog交给其加工过滤，而eventSink又要把处理后的数据交给eventStore进行存储。
     * 因此依赖关系如下：eventStore—>eventSink—>eventParser ，启动的时候也要按照这个顺序启动。
     */
    @Override
    public void start() {
        super.start();
        if (!metaManager.isStart()) {
            metaManager.start();
        }

        if (!alarmHandler.isStart()) {
            alarmHandler.start();
        }

        if (!eventStore.isStart()) {
            eventStore.start();
        }

        if (!eventSink.isStart()) {
            eventSink.start();
        }

        // http://static.tianshouzhi.com/ueditor/upload/image/20180506/1525619555225049221.png
        if (!eventParser.isStart()) {
            // eventParser在启动之前，需要先启动CanalLogPositionManager和CanalHAController。
            // CanalLogPositionManager: mysql在主从同步过程中，要求slave自己维护binlog的消费进度信息。canal伪装成slave，因此也要维护这样的信息。
            // 在slave机器的data目录下，都会有一个master.info文件，这个文件的作用就是存储主库的消费binlog解析进度信息。
            beforeStartEventParser(eventParser); //启动前执行一些操作
            eventParser.start();
            afterStartEventParser(eventParser); //启动后执行一些操作
        }
        logger.info("start successful....");
    }

    /**
     * 在停止的时候，实际上就是停止内部的各个模块，模块停止的顺序与start方法刚好相反
     */
    @Override
    public void stop() {
        super.stop();
        logger.info("stop CannalInstance for {}-{} ", new Object[] { canalId, destination });

        if (eventParser.isStart()) {
            beforeStopEventParser(eventParser); // 停止前执行一些操作
            eventParser.stop();
            afterStopEventParser(eventParser); // 停止后执行一些操作
        }

        if (eventSink.isStart()) {
            eventSink.stop();
        }

        if (eventStore.isStart()) {
            eventStore.stop();
        }

        if (metaManager.isStart()) {
            metaManager.stop();
        }

        if (alarmHandler.isStart()) {
            alarmHandler.stop();
        }

        logger.info("stop successful....");
    }

    /**
     *  beforeStartEventParser方法的作用是eventParser前做的一些特殊处理。
     *  首先会判断eventParser的类型是否是GroupEventParser，这是为了处理分库分表的情况。
     *  如果是，循环其包含的所有CanalEventParser，依次调用startEventParserInternal方法；
     *  否则直接调用com.alibaba.otter.canal.instance.core.AbstractCanalInstance#beforeStartEventParser
     */
    protected void beforeStartEventParser(CanalEventParser eventParser) {
        // 1、判断eventParser的类型是否是GroupEventParser
        boolean isGroup = (eventParser instanceof GroupEventParser);

        // 2、如果是GroupEventParser，则循环启动其内部包含的每一个CanalEventParser，依次调用startEventParserInternal方法
        if (isGroup) {
            // 处理group的模式
            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                startEventParserInternal(singleEventParser, true);
            }
        }
        // 如果不是，说明是一个普通的CanalEventParser，直接调用startEventParserInternal方法
        else {
            // 针对单个CanalEventParser，都是通过调用startEventParserInternal来启动的，其内部会启动CanalLogPositionManager和CanalHAController。
            startEventParserInternal(eventParser, false); //
        }
    }

    /**
     * 在eventParser启动后，会调用afterStartEventParser方法。
     * 这个方法内部主要是通过metaManager读取一下历史订阅过这个CanalInstance的客户端信息，然后更新一下filter。
     */
    protected void afterStartEventParser(CanalEventParser eventParser) {
        // 读取一下历史订阅的filter信息
        List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
        for (ClientIdentity clientIdentity : clientIdentitys) {
            // 更新filter
            subscribeChange(clientIdentity); //
        }
    }

    // around event parser
    protected void beforeStopEventParser(CanalEventParser eventParser) {
        // noop
    }

    protected void afterStopEventParser(CanalEventParser eventParser) {

        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup) {
            // 处理group的模式
            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                stopEventParserInternal(singleEventParser);
            }
        } else {
            stopEventParserInternal(eventParser);
        }
    }

    /**
     * 初始化单个eventParser，不需要考虑group
     */
    protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
        // 1 、启动 CanalLogPositionManager
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (!logPositionManager.isStart()) {
                logPositionManager.start();
            }
        }

        // 2 、启动 CanalHAController
        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            CanalHAController haController = mysqlEventParser.getHaController();

            if (haController instanceof HeartBeatHAController) {
                ((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
            }

            if (!haController.isStart()) {
                haController.start();
            }

        }
    }

    protected void stopEventParserInternal(CanalEventParser eventParser) {
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (logPositionManager.isStart()) {
                logPositionManager.stop();
            }
        }

        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            CanalHAController haController = mysqlEventParser.getHaController();
            if (haController.isStart()) {
                haController.stop();
            }
        }
    }

    // ==================getter==================================
    @Override
    public String getDestination() {
        return destination;
    }

    @Override
    public CanalEventParser getEventParser() {
        return eventParser;
    }

    @Override
    public CanalEventSink getEventSink() {
        return eventSink;
    }

    @Override
    public CanalEventStore getEventStore() {
        return eventStore;
    }

    @Override
    public CanalMetaManager getMetaManager() {
        return metaManager;
    }

    @Override
    public CanalAlarmHandler getAlarmHandler() {
        return alarmHandler;
    }

    @Override
    public CanalMQConfig getMqConfig() {
        return mqConfig;
    }
}
